#include "ThreadPool.h"
#include <iostream>
#include <unistd.h>
#include <string.h>
using namespace std;

ThreadPool::ThreadPool(int min, int max)
{
    //任务队列实例化
    taskQ = new TaskQueue();
    do{
        //初始化线程池
        minNum = min;
        maxNum = max;
        busyNum = 0;
        liveNum = minNum;
        shutdown = 0;

        //根据线程池的最大容量给线程数组分配内存
        threadIDs = new pthread_t[maxNum];
        if (threadIDs == nullptr)
            {
                cout << "new pthread_t[maxNum] failed...." << endl;
                break;
            }
        //初始化
        memset(threadIDs,0,sizeof(pthread_t) * maxNum);
        //初始化互斥锁,条件变量
        if(pthread_mutex_init(&mutexPool,NULL) != 0|| pthread_cond_init(&notEmpty,NULL) != 0){
            cout<<"init mutex or condition failed..."<<endl;
            break;
        }

        //创建线程
        //根据线程池的最小容量,创建线程
        for(int i = 0;i<minNum;i++){
            /*
            worker工作函数必须是全局函数或者静态成员函数
            因为如果worker是成员函数,那么如果类没有创建实例,就不存在这个函数,没有这个地址,pthread_create就没办法绑定
            然后静态函数内部不能访问非静态数据,所以通过传递的this指针来访问本类的数据
            */
            pthread_create(&threadIDs[i],NULL,worker,this);    
            cout<<"create child thread , ID: "<<to_string(threadIDs[i])<<endl;
        }
        //创建管理者线程,1个
        pthread_create(&managerID,NULL,manager,this);
         cout<<"create manager thread , ID: "<<to_string(managerID)<<endl;

        return; //执行成功,则不需要释放资源

    }while(false);
   
   //初始化失败直接释放资源
   if (taskQ)
        delete taskQ;

    if (threadIDs) 
        delete[] threadIDs;

}

ThreadPool::~ThreadPool()
{
    shutdown = true;
    // 销毁管理者线程
    pthread_join(managerID, NULL);  //线程传递出来的数据
    // 唤醒所有消费者线程
    for (int i = 0; i < liveNum; ++i)       //最后没有任务做了,工作线程都是阻塞的,所以有多少个线程就唤醒多少个
    {
        pthread_cond_signal(&notEmpty);
    }
    if (taskQ)
        delete taskQ;

    if (threadIDs) 
        delete[] threadIDs;

    pthread_mutex_destroy(&mutexPool);
    pthread_cond_destroy(&notEmpty);

}

void ThreadPool::addTask(Task task)
{
     if (shutdown)
    {
        return;
    }
    // 添加任务，不需要加锁，任务队列中有锁
    taskQ->addTask(task);
    // 唤醒工作的线程
    pthread_cond_signal(&notEmpty);

}

int ThreadPool::getBusyNum()
{
    int t_busyNum = 0;
    pthread_mutex_lock(&mutexPool);
    t_busyNum = busyNum;
    pthread_mutex_unlock(&mutexPool);
    return t_busyNum;

}

int ThreadPool::getAliveNum()
{
    int threadNum = 0;
    pthread_mutex_lock(&mutexPool);
    threadNum = liveNum;
    pthread_mutex_unlock(&mutexPool);
    return threadNum;

}

void *ThreadPool::worker(void *arg)
{
    ThreadPool* pool = static_cast<ThreadPool*>(arg);
    // 一直不停的工作
    while (true)
    {
        // 访问任务队列(共享资源)加锁
        pthread_mutex_lock(&pool->mutexPool);
        // 判断任务队列是否为空, 如果为空,工作线程阻塞
        while (pool->taskQ->taskNumber() == 0 && !pool->shutdown)
        {
            cout << "thread " << to_string(pthread_self()) << " waiting..." << endl;
            // 阻塞线程
            pthread_cond_wait(&pool->notEmpty, &pool->mutexPool);

            // 解除阻塞之后, 判断是否要销毁线程
            if (pool->exitNum > 0)
            {
                pool->exitNum--;        //会销毁两个
                if (pool->liveNum > pool->minNum)   //销毁过程中要判定 保证活着的线程个数 > 线程池最小容量
                {
                    pool->liveNum--;
                    pthread_mutex_unlock(&pool->mutexPool);
                    pool->threadExit(); //线程退出
                }
            }
        }
        // 判断线程池是否被关闭了
        if (pool->shutdown)
        {
            pthread_mutex_unlock(&pool->mutexPool);
            pool->threadExit(); //线程自毁
        }

        // 从任务队列中取出一个任务
        Task task = pool->taskQ->getTask();
        // 工作的线程+1
        pool->busyNum++;
        // 线程池解锁
        pthread_mutex_unlock(&pool->mutexPool);
        // 执行任务
        cout << "thread " << to_string(pthread_self()) << " start working..." << endl;
        task.function(task.arg);
        //delete task.arg;
        task.arg = nullptr;

        // 任务处理结束
        cout << "thread " << to_string(pthread_self()) << " end working..."<<endl;
        pthread_mutex_lock(&pool->mutexPool);
        pool->busyNum--;
        pthread_mutex_unlock(&pool->mutexPool);
    }
    return nullptr;
}

void *ThreadPool::manager(void *arg)
{
    ThreadPool* pool = static_cast<ThreadPool*>(arg);
    // 如果线程池没有关闭, 就一直检测
    while (!pool->shutdown)
    {
        // 每隔5s检测一次
        sleep(5);
        // 取出线程池中的任务数和线程数量
        //  取出工作的线程池数量
        pthread_mutex_lock(&pool->mutexPool);
        int t_queueSize = pool->taskQ->taskNumber();
        int t_liveNum = pool->liveNum;
        int t_busyNum = pool->busyNum;
        pthread_mutex_unlock(&pool->mutexPool);

        //线程不够用时创建线程
        // 当前任务个数>存活的线程数 && 存活的线程数<最大线程个数
        if (t_queueSize > t_liveNum && t_liveNum < pool->maxNum)
        {
            // 线程池加锁
            pthread_mutex_lock(&pool->mutexPool);
            int num = 0;
            for (int i = 0; i < pool->maxNum && num < NUMBER
                && pool->liveNum < pool->maxNum; ++i)
            {
                if (pool->threadIDs[i] == 0)
                {
                    pthread_create(&pool->threadIDs[i], NULL, worker, pool);
                    num++;      //只会创建2个线程
                    pool->liveNum++;    //存活线程数加一
                }
            }
            pthread_mutex_unlock(&pool->mutexPool);
        }

        // 销毁多余的线程
        // 忙线程*2 < 存活的线程数目 && 存活的线程数 > 最小线程数量
        if (pool->busyNum * 2 < pool->liveNum && pool->liveNum > pool->minNum)
        {
            pthread_mutex_lock(&pool->mutexPool);
            pool->exitNum = NUMBER; //要销毁的线程个数
            pthread_mutex_unlock(&pool->mutexPool);
            for (int i = 0; i < pool->exitNum; ++i)
            {
                pthread_cond_signal(&pool->notEmpty); //唤醒阻塞在条件变量上的线程
            }
        }
    }
    return nullptr;
}

void ThreadPool::threadExit()
{
    /*pthread_self()返回的是posix定义的线程ID(用户态线程ID)，
    它只是用来区分某个进程中不同的线程，当一个线程退出后，新创建的线程可以复用原来的id
    */
    pthread_t tid = pthread_self(); //获取线程id
    for (int i = 0; i < maxNum; ++i)
    {
        if (threadIDs[i] == tid)
        {
            cout << "threadExit() function: thread " 
                << to_string(pthread_self()) << " exiting..." << endl;
            threadIDs[i] = 0;
            break;
        }
    }
    pthread_exit(NULL);

}
